package core_sql.day04

import java.sql.{Connection, DriverManager, PreparedStatement}

import util.FileUtil
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}



/**
  * Created by root on 2019/3/11.
  * 广播变量
  *
  */
object IPLocation {
  /**
    * ip转换为long型的ip
    * @param ip
    * @return
    */
  def ip2Long(ip:String):Long = {
    val fragment: Array[String] = ip.split("\\.")
    var ipNum = 0L
    for( i <- 0 until fragment.length){
      ipNum = fragment(i).toInt | ipNum << 8L
    }
    ipNum
  }

  /**
    * 二分查找
    * @param lines
    * @param ip
    * @return
    */
  def binarySearch(lines:Array[(Long,Long,String)],ip:Long) :Int ={
    var low = 0;
    var high = lines.length - 1
    while(low <= high){
      val middle = (low+high) /2
      if((ip >=lines(middle)._1) && ip <=lines(middle)._2)
        return middle
      if(ip < lines(middle)._1)
        high=middle-1
      else
        low = middle +1
    }
    -1
  }

  /**
    * 数据写入到数据库中
    * @param it
    */
  def data2MySQL(it:Iterator[(String,Int)]) :Unit ={
    var conn :Connection = null
    var pstm:PreparedStatement = null
    try{
      conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/spark?characterEncoding=UTF-8","root","123456")
      pstm = conn.prepareStatement("insert into access_log values(?,?)")
      //将数据写入，将一个分区中的数据写完后再关闭连接
      it.foreach(tp => {
        pstm.setString(1,tp._1)
        pstm.setInt(2,tp._2)
        pstm.executeUpdate()
      })
    } catch {
      case e:Exception => println(e.printStackTrace())
    }finally {
      if(pstm != null){
        pstm.close()
      }
      if(conn !=null) conn.close()
    }

  }


  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("IPLocation").setMaster("local[4]")

    val sc = new SparkContext(conf)
    //前期准备工作（下发一个job任务，收集全部的IP规则到Driver）

    val ipRuleLines: RDD[String] = sc.textFile(FileUtil.IP_LOG)
    //切割收集到driver端
    val ipRules: Array[(Long, Long, String)] = ipRuleLines.map(line => {
      val fields = line.split("[|]")
      val startNum = fields(2).toLong
      val endNum = fields(3).toLong
      val province = fields(6)
      (startNum, endNum, province)
    }).collect()

    //将全部的IP规则通过网络让Driver方送给属于该任务的Executor中，其实是缓存在Executor进程的内存中了
    val ipRuleBroadCast: Broadcast[Array[(Long, Long, String)]] = sc.broadcast(ipRules)

    //读取访问日志
    val accessLines: RDD[String] = sc.textFile(FileUtil.ACCESS_LOG)

    //整理数据，聚合数据
    val reduced:RDD[(String,Int)] = accessLines.map(line => {
      val fields: Array[String] = line.split("\\|")
      val ip = fields(1)
      //匹配ip规则
      val ipNum: Long = ip2Long(ip)
      //在Executor中通过广播变量的引用获得缓存的全部IP规则
      
      val cachedIpRules = ipRuleBroadCast.value
      //使用二分法查找匹配
      val index:Int = binarySearch(cachedIpRules,ipNum)
      val province: String = cachedIpRules(index)._3
      (province,1)
    }).reduceByKey(_+_)

    //将数据写到数据库中
    reduced.foreachPartition(data2MySQL(_))

    sc.stop()
  }

}
